/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.broker.processor;

import com.alibaba.rocketmq.broker.BrokerController;
import com.alibaba.rocketmq.broker.client.ConsumerGroupInfo;
import com.alibaba.rocketmq.broker.longpolling.PullRequest;
import com.alibaba.rocketmq.broker.mqtrace.ConsumeMessageContext;
import com.alibaba.rocketmq.broker.mqtrace.ConsumeMessageHook;
import com.alibaba.rocketmq.broker.pagecache.ManyMessageTransfer;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.TopicConfig;
import com.alibaba.rocketmq.common.TopicFilterType;
import com.alibaba.rocketmq.common.constant.LoggerName;
import com.alibaba.rocketmq.common.constant.PermName;
import com.alibaba.rocketmq.common.filter.FilterAPI;
import com.alibaba.rocketmq.common.help.FAQUrl;
import com.alibaba.rocketmq.common.message.MessageDecoder;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.ResponseCode;
import com.alibaba.rocketmq.common.protocol.header.PullMessageRequestHeader;
import com.alibaba.rocketmq.common.protocol.header.PullMessageResponseHeader;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.alibaba.rocketmq.common.protocol.heartbeat.SubscriptionData;
import com.alibaba.rocketmq.common.protocol.topic.OffsetMovedEvent;
import com.alibaba.rocketmq.common.subscription.SubscriptionGroupConfig;
import com.alibaba.rocketmq.common.sysflag.PullSysFlag;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;
import com.alibaba.rocketmq.remoting.common.RemotingUtil;
import com.alibaba.rocketmq.remoting.exception.RemotingCommandException;
import com.alibaba.rocketmq.remoting.netty.NettyRequestProcessor;
import com.alibaba.rocketmq.remoting.protocol.RemotingCommand;
import com.alibaba.rocketmq.store.GetMessageResult;
import com.alibaba.rocketmq.store.MessageExtBrokerInner;
import com.alibaba.rocketmq.store.PutMessageResult;
import com.alibaba.rocketmq.store.config.BrokerRole;
import io.netty.channel.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.List;
import java.util.Map;

/**
 * 拉消息请求处理
 * 
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-26
 */
public class PullMessageProcessor implements NettyRequestProcessor {
	private static final Logger log = LoggerFactory.getLogger(LoggerName.BrokerLoggerName);

	private final BrokerController brokerController;

	public PullMessageProcessor(final BrokerController brokerController) {
		this.brokerController = brokerController;
	}

	@Override
	public RemotingCommand processRequest(final ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
		return this.processRequest(ctx.channel(), request, true);
	}

	public void excuteRequestWhenWakeup(final Channel channel, final RemotingCommand request) throws RemotingCommandException {
		// 首先创建一个线程
		Runnable run = new Runnable() {
			@Override
			public void run() {
				try {
					// 其中入参brokerAllowSuspend等于false，表示若未拉取到消息，则不再采用该线程的补偿机制了；然后继续遍历下一个PullRequest对象；
					final RemotingCommand response = PullMessageProcessor.this.processRequest(channel, request, false);

					if (response != null) {
						response.setOpaque(request.getOpaque());
						response.markResponseType();
						try {
							channel.writeAndFlush(response).addListener(new ChannelFutureListener() {
								@Override
								public void operationComplete(ChannelFuture future) throws Exception {
									if (!future.isSuccess()) {
										log.error("processRequestWrapper response to " + future.channel().remoteAddress() + " failed", future.cause());
										log.error(request.toString());
										log.error(response.toString());
									}
								}
							});
						} catch (Throwable e) {
							log.error("processRequestWrapper process request over, but response failed", e);
							log.error(request.toString());
							log.error(response.toString());
						}
					}
				} catch (RemotingCommandException e1) {
					log.error("excuteRequestWhenWakeup run", e1);
				}
			}
		};
		// 将该线程放入线程池中
		this.brokerController.getPullMessageExecutor().submit(run);
	}

	private void generateOffsetMovedEvent(final OffsetMovedEvent event) {
		try {
			MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
			msgInner.setTopic(MixAll.OFFSET_MOVED_EVENT);
			msgInner.setTags(event.getConsumerGroup());
			msgInner.setDelayTimeLevel(0);
			msgInner.setKeys(event.getConsumerGroup());
			msgInner.setBody(event.encode());
			msgInner.setFlag(0);
			msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
			msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(TopicFilterType.SINGLE_TAG, msgInner.getTags()));

			msgInner.setQueueId(0);
			msgInner.setSysFlag(0);
			msgInner.setBornTimestamp(System.currentTimeMillis());
			msgInner.setBornHost(RemotingUtil.string2SocketAddress(this.brokerController.getBrokerAddr()));
			msgInner.setStoreHost(msgInner.getBornHost());

			msgInner.setReconsumeTimes(0);
			// 最后调用MessageStore.putMessage(MessageExtBrokerInner msg)方法将该消息存入CommitLog中；
			PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
		} catch (Exception e) {
			log.warn(String.format("generateOffsetMovedEvent Exception, %s", event.toString()), e);
		}
	}

	private RemotingCommand processRequest(final Channel channel, RemotingCommand request, boolean brokerAllowSuspend) throws RemotingCommandException {
		RemotingCommand response = RemotingCommand.createResponseCommand(PullMessageResponseHeader.class);
		final PullMessageResponseHeader responseHeader = (PullMessageResponseHeader) response.readCustomHeader();
		final PullMessageRequestHeader requestHeader = (PullMessageRequestHeader) request.decodeCommandCustomHeader(PullMessageRequestHeader.class);

		// 由于使用sendfile，所以必须要设置
		response.setOpaque(request.getOpaque());

		if (log.isDebugEnabled()) {
			log.debug("receive PullMessage request command, " + request);
		}

		// 检查Broker权限
		if (!PermName.isReadable(this.brokerController.getBrokerConfig().getBrokerPermission())) {
			response.setCode(ResponseCode.NO_PERMISSION);
			response.setRemark("the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1() + "] pulling message is forbidden");
			return response;
		}

		// 确保订阅组存在
		SubscriptionGroupConfig subscriptionGroupConfig = this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(requestHeader.getConsumerGroup());
		if (null == subscriptionGroupConfig) {
			// 若没有找到该Consumer的订阅信息并且Broker不允许自动创建订阅组,则返回响应消息，消息代码为SUBSCRIPTION_GROUP_NOT_EXIST
			response.setCode(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST);
			response.setRemark("subscription group not exist, " + requestHeader.getConsumerGroup() + " " + FAQUrl.suggestTodo(FAQUrl.SUBSCRIPTION_GROUP_NOT_EXIST));
			return response;
		}

		// 这个订阅组是否可以消费消息
		// 若找到该Consumer的订阅信息则检查该订阅组是否可以消费消息，即SubscriptionGroupConfig.consumeEnable变量是否为true
		if (!subscriptionGroupConfig.isConsumeEnable()) {
			// 若不能则直接返回响应消息，消息代码为NO_PERMISSION
			response.setCode(ResponseCode.NO_PERMISSION);
			response.setRemark("subscription group no permission, " + requestHeader.getConsumerGroup());
			return response;
		}

		/**
		 * 拉取消息PullMessageRequestHeader对象的sysFlag标志位，第一位表示commitOffsetEnable，第二位表示suspend；第三位表示subscription；第四位表示classFilterMode，
		 * 这些字段都是有应用端设置之后，由buildSysFlag(final boolean commitOffset, final boolean suspend,final boolean subscription, final boolean classFilter)方法生成的sysFlag值；
		 */
		final boolean hasSuspendFlag = PullSysFlag.hasSuspendFlag(requestHeader.getSysFlag());
		final boolean hasCommitOffsetFlag = PullSysFlag.hasCommitOffsetFlag(requestHeader.getSysFlag());
		final boolean hasSubscriptionFlag = PullSysFlag.hasSubscriptionFlag(requestHeader.getSysFlag());

		final long suspendTimeoutMillisLong = hasSuspendFlag ? requestHeader.getSuspendTimeoutMillis() : 0;

		// 检查topic是否存在
		// 根据请求消息中的topic值从TopicConfigManager.topicConfigTable中获取TopicConfig对象，
		TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
		if (null == topicConfig) {
			// 若为空则表示consumer消费的topic不存在，直接返回响应消息，消息代码为TOPIC_NOT_EXIST；
			log.error("the topic " + requestHeader.getTopic() + " not exist, consumer: " + RemotingHelper.parseChannelRemoteAddr(channel));
			response.setCode(ResponseCode.TOPIC_NOT_EXIST);
			response.setRemark("topic[" + requestHeader.getTopic() + "] not exist, apply first please!" + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
			return response;
		}

		// 若不为空则首先
		// 检查topicConfig是否有读的权限
		if (!PermName.isReadable(topicConfig.getPerm())) {
			response.setCode(ResponseCode.NO_PERMISSION);
			response.setRemark("the topic[" + requestHeader.getTopic() + "] pulling message is forbidden");
			return response;
		}

		// 检查队列有效性
		// 然后检查请求消费的队列ID是否有效（即要大于0并且小于读队列总数），
		if (requestHeader.getQueueId() < 0 || requestHeader.getQueueId() >= topicConfig.getReadQueueNums()) {
			String errorInfo = "queueId[" + requestHeader.getQueueId() + "] is illagal,Topic :" + requestHeader.getTopic() + " topicConfig.readQueueNums: " + topicConfig.getReadQueueNums() + " consumer: " + channel.remoteAddress();
			log.warn(errorInfo);
			// 若无效则返回响应消息，消息代码为SYSTEM_ERROR；
			response.setCode(ResponseCode.SYSTEM_ERROR);
			response.setRemark(errorInfo);
			return response;
		}

		// 订阅关系处理
		SubscriptionData subscriptionData = null;
		// 检查PullMessageRequestHeader对象的sysFlag标志位的第三位subscription标识
		if (hasSubscriptionFlag) {
			try {
				// 若为true，则直接用请求消息中的consumerGroup、topic、subscription构建SubscriptionData对象
				subscriptionData = FilterAPI.buildSubscriptionData(requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getSubscription());
			} catch (Exception e) {
				log.warn("parse the consumer's subscription[{}] failed, group: {}", requestHeader.getSubscription(), //
						requestHeader.getConsumerGroup());
				response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
				response.setRemark("parse the consumer's subscription failed");
				return response;
			}
		} else {
			// 得到对应的consumer group的consumerGroupInfo
			ConsumerGroupInfo consumerGroupInfo = this.brokerController.getConsumerManager().getConsumerGroupInfo(requestHeader.getConsumerGroup());
			if (null == consumerGroupInfo) {
				log.warn("the consumer's group info not exist, group: {}", requestHeader.getConsumerGroup());
				response.setCode(ResponseCode.SUBSCRIPTION_NOT_EXIST);
				response.setRemark("the consumer's group info not exist" + FAQUrl.suggestTodo(FAQUrl.SAME_GROUP_DIFFERENT_TOPIC));
				return response;
			}

			if (!subscriptionGroupConfig.isConsumeBroadcastEnable() //
					&& consumerGroupInfo.getMessageModel() == MessageModel.BROADCASTING) {
				response.setCode(ResponseCode.NO_PERMISSION);
				response.setRemark("the consumer group[" + requestHeader.getConsumerGroup() + "] can not consume by broadcast way");
				return response;
			}

			subscriptionData = consumerGroupInfo.findSubscriptionData(requestHeader.getTopic());
			if (null == subscriptionData) {
				log.warn("the consumer's subscription not exist, group: {}", requestHeader.getConsumerGroup());
				response.setCode(ResponseCode.SUBSCRIPTION_NOT_EXIST);
				response.setRemark("the consumer's subscription not exist" + FAQUrl.suggestTodo(FAQUrl.SAME_GROUP_DIFFERENT_TOPIC));
				return response;
			}

			// 判断Broker的订阅关系版本是否最新
			if (subscriptionData.getSubVersion() < requestHeader.getSubVersion()) {
				log.warn("the broker's subscription is not latest, group: {} {}", requestHeader.getConsumerGroup(), subscriptionData.getSubString());
				response.setCode(ResponseCode.SUBSCRIPTION_NOT_LATEST);
				response.setRemark("the consumer's subscription not latest");
				return response;
			}
		}

		// 读取消息
		final GetMessageResult getMessageResult = this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getQueueOffset(), requestHeader.getMaxMsgNums(), subscriptionData);
		if (getMessageResult != null) {
			response.setRemark(getMessageResult.getStatus().name());
			// 用DefaultMessageStore.getMessage方法的返回值GetMessageResult对象中的参数值来初始化PullMessageResponseHeader对象
			responseHeader.setNextBeginOffset(getMessageResult.getNextBeginOffset());
			responseHeader.setMinOffset(getMessageResult.getMinOffset());
			responseHeader.setMaxOffset(getMessageResult.getMaxOffset());

			// 消费较慢，重定向到另外一台机器
			if (getMessageResult.isSuggestPullingFromSlave()) {
				// 若GetMessageResult.suggestPullingFromSlave等于true，
				// 即表示该Broker消费很慢，存在消息堆积后，设置suggestWhichBrokerId等于1，即将Consumer的消费请求重定向到另外一台Slave机器。
				responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
			}
			// 消费正常，按照订阅组配置重定向
			else {
				// 若为false则将suggestWhichBrokerId仍然设置为当前Broker的Id；
				responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getBrokerId());
			}

			// 访问消息返回的状态码
			/**
			 * 
			// 找到消息
			FOUND,
			// offset正确，但是过滤后没有匹配的消息
			NO_MATCHED_MESSAGE,
			// offset正确，但是物理队列消息正在被删除
			MESSAGE_WAS_REMOVING,
			// offset正确，但是从逻辑队列没有找到，可能正在被删除
			OFFSET_FOUND_NULL,
			// offset错误，严重溢出
			OFFSET_OVERFLOW_BADLY,
			// offset错误，溢出1个
			OFFSET_OVERFLOW_ONE,
			// offset错误，太小了
			OFFSET_TOO_SMALL,
			// 没有对应的逻辑队列
			NO_MATCHED_LOGIC_QUEUE,
			// 队列中一条消息都没有
			NO_MESSAGE_IN_QUEUE,
			 */
			switch (getMessageResult.getStatus()) {
				// 找到消息
				case FOUND :
					response.setCode(ResponseCode.SUCCESS);

					// 消息轨迹：记录客户端拉取的消息记录（不表示消费成功）
					// 然后检查Broker是否设置了回调类PullMessageProcessor.ConsumeMessageHookList；
					if (this.hasConsumeMessageHook()) {
						// 执行hook
						ConsumeMessageContext context = new ConsumeMessageContext();
						context.setConsumerGroup(requestHeader.getConsumerGroup());
						context.setTopic(requestHeader.getTopic());
						context.setClientHost(RemotingHelper.parseChannelRemoteAddr(channel));
						context.setStoreHost(this.brokerController.getBrokerAddr());
						context.setQueueId(requestHeader.getQueueId());

						final SocketAddress storeHost = new InetSocketAddress(brokerController.getBrokerConfig().getBrokerIP1(), brokerController.getNettyServerConfig().getListenPort());
						Map<String, Long> messageIds = this.brokerController.getMessageStore().getMessageIds(requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getQueueOffset(), requestHeader.getQueueOffset() + getMessageResult.getMessageCount(), storeHost);
						context.setMessageIds(messageIds);
						context.setBodyLength(getMessageResult.getBufferTotalSize() / getMessageResult.getMessageCount());
						// 若设置了则执行该回调类中的consumeMessageBefore方法；
						this.executeConsumeMessageHookBefore(context);
					}

					break;
				// offset正确，但是物理队列消息正在被删除
				case MESSAGE_WAS_REMOVING :
					response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
					break;
				// 这两个返回值都表示服务器暂时没有这个队列，应该立刻将客户端Offset重置为0
				case NO_MATCHED_LOGIC_QUEUE :
					// 队列中一条消息都没有
				case NO_MESSAGE_IN_QUEUE :
					// 当请求消息中的读取开始位置offset值不等于0时
					if (0 != requestHeader.getQueueOffset()) {
						// 将RemotingCommand的code设置为PULL_OFFSET_MOVED
						// （拉消息请求的Offset不合法，太小或太大）
						response.setCode(ResponseCode.PULL_OFFSET_MOVED);

						// XXX: warn and notify me
						log.info("the broker store no queue data, fix the request offset {} to {}, Topic: {} QueueId: {} Consumer Group: {}", //
								requestHeader.getQueueOffset(), //
								getMessageResult.getNextBeginOffset(), //
								requestHeader.getTopic(), //
								requestHeader.getQueueId(), //
								requestHeader.getConsumerGroup()//
						);
					} else {
						// 当请求消息中的读取开始位置offset值等于0时将RemotingCommand的code设置为PULL_NOT_FOUND；
						response.setCode(ResponseCode.PULL_NOT_FOUND);
					}
					break;
				// offset正确，但是过滤后没有匹配的消息
				case NO_MATCHED_MESSAGE :
					response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
					break;
				// offset正确，但是从逻辑队列没有找到，可能正在被删除
				// 未获取到consumequeue数据
				case OFFSET_FOUND_NULL :
					response.setCode(ResponseCode.PULL_NOT_FOUND);
					break;
				// offset错误，严重溢出
				// 请求参数offset大于consumequeue中最大物理偏移量
				case OFFSET_OVERFLOW_BADLY :
					response.setCode(ResponseCode.PULL_OFFSET_MOVED);
					// XXX: warn and notify me
					log.info("the request offset: " + requestHeader.getQueueOffset() + " over flow badly, broker max offset: " + getMessageResult.getMaxOffset() + ", consumer: " + channel.remoteAddress());
					break;
				// offset错误，溢出1个
				// 请求参数offset等于consumequeue中最大物理偏移量
				case OFFSET_OVERFLOW_ONE :
					response.setCode(ResponseCode.PULL_NOT_FOUND);
					break;
				// offset错误，太小了
				// 请求参数offset小于consumequeue中最小物理偏移量
				case OFFSET_TOO_SMALL :
					response.setCode(ResponseCode.PULL_OFFSET_MOVED);
					// XXX: warn and notify me
					log.info("the request offset: " + requestHeader.getQueueOffset() + " too small, broker min offset: " + getMessageResult.getMinOffset() + ", consumer: " + channel.remoteAddress());
					break;
				default :
					assert false;
					break;
			}

			// 11、检查返回对象RemotingCommand的code值
			switch (response.getCode()) {
				case ResponseCode.SUCCESS :
					// 统计
					this.brokerController.getBrokerStatsManager().incGroupGetNums(requestHeader.getConsumerGroup(), requestHeader.getTopic(), getMessageResult.getMessageCount());

					this.brokerController.getBrokerStatsManager().incGroupGetSize(requestHeader.getConsumerGroup(), requestHeader.getTopic(), getMessageResult.getBufferTotalSize());

					this.brokerController.getBrokerStatsManager().incBrokerGetNums(getMessageResult.getMessageCount());

					try {
						// 初始化ManyMessageTransfer对象，该对象实现了Netty的FileRegion接口
						FileRegion fileRegion = new ManyMessageTransfer(response.encodeHeader(getMessageResult.getBufferTotalSize()), getMessageResult);
						// 调用Channel.WriteAndFlush方法将该ManyMessageTransfer对象直接写入Channel中将获取到的数据返回给调用者
						// 并添加一个渠道监听器ChannelFutureListener
						channel.writeAndFlush(fileRegion).addListener(new ChannelFutureListener() {
							// 该监听器实现了operationComplete方法，当消息发送完成后调用此监听器的该方法，
							@Override
							public void operationComplete(ChannelFuture future) throws Exception {
								// 在方法中调用GetMessageResult对象的release方法，释放资源
								getMessageResult.release();
								if (!future.isSuccess()) {
									log.error("transfer many message by pagecache failed, " + channel.remoteAddress(), future.cause());
								}
							}
						});
					} catch (Throwable e) {
						log.error("", e);
						getMessageResult.release();
					}

					response = null;
					break;
				case ResponseCode.PULL_NOT_FOUND :
					// 允许长轮询
					if (brokerAllowSuspend && hasSuspendFlag) {
						// 若是支持长轮询则设置该对象的timeoutMillis变量等于请求消息的suspendTimeoutMillis值
						long pollingTimeMills = suspendTimeoutMillisLong;
						// 判断该Broker是否支持长轮询(BrokerConfig.longPollingEnable设置，默认为true)
						if (!this.brokerController.getBrokerConfig().isLongPollingEnable()) {
							// 支持短轮询，设置timeoutMillis变量等于BrokerConfig.shortPollingTimeMills的值
							pollingTimeMills = this.brokerController.getBrokerConfig().getShortPollingTimeMills();
						}

						PullRequest pullRequest = new PullRequest(request, channel, pollingTimeMills, this.brokerController.getMessageStore().now(), requestHeader.getQueueOffset());
						// 将PullRequest对象存入PullRequestHoldService.pullRequestTable变量中，由PullRequestHoldService服务线程定期检查是否有符合要求的数据
						this.brokerController.getPullRequestHoldService().suspendPullRequest(requestHeader.getTopic(), requestHeader.getQueueId(), pullRequest);
						response = null;
						break;
					}

					// 向Consumer返回应答
				case ResponseCode.PULL_RETRY_IMMEDIATELY :
					break;
				case ResponseCode.PULL_OFFSET_MOVED :
					// 当该broker角色是主用或者BrokerConfig.offsetCheckInSlave（是否需要纠正位点，默认为false）等于true时，
					if (this.brokerController.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE || this.brokerController.getBrokerConfig().isOffsetCheckInSlave()) {
						// 初始化MessageQueue对象
						MessageQueue mq = new MessageQueue();
						mq.setTopic(requestHeader.getTopic());
						mq.setQueueId(requestHeader.getQueueId());
						mq.setBrokerName(this.brokerController.getBrokerConfig().getBrokerName());

						// 初始化OffsetMovedEvent对象
						OffsetMovedEvent event = new OffsetMovedEvent();
						event.setConsumerGroup(requestHeader.getConsumerGroup());
						event.setMessageQueue(mq);
						event.setOffsetRequest(requestHeader.getQueueOffset());
						event.setOffsetNew(getMessageResult.getNextBeginOffset());

						// 构建topic值为"OFFSET_MOVED_EVENT"的消息对象MessageExtBrokerInner，其中消息内容为OffsetMovedEvent对象的序列化
						this.generateOffsetMovedEvent(event);
					} else {
						// 当该broker角色是备用或者BrokerConfig.offsetCheckInSlave（是否需要纠正位点，默认为false）等于false时
						responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getBrokerId());
						response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
					}

					log.warn("PULL_OFFSET_MOVED:topic={}, groupId={}, clientId={}, offset={}, suggestBrokerId={}", requestHeader.getTopic(), requestHeader.getConsumerGroup(), requestHeader.getQueueOffset(), responseHeader.getSuggestWhichBrokerId());
					break;
				default :
					assert false;
			}
		} else {
			response.setCode(ResponseCode.SYSTEM_ERROR);
			response.setRemark("store getMessage return null");
		}

		// 存储Consumer消费进度
		boolean storeOffsetEnable = brokerAllowSuspend; // 说明是首次调用，相对于长轮询通知
		storeOffsetEnable = storeOffsetEnable && hasCommitOffsetFlag; // 说明Consumer设置了标志位
		storeOffsetEnable = storeOffsetEnable // 只有Master支持存储offset
				&& this.brokerController.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE;
		if (storeOffsetEnable) {
			/**
			 * 若broker是主用，
			 * 并且允许长轮询（入参brokerAllowSuspend=true）
			 * 并且SysFlag的commitoffset标记位为true
			 */
			// 将请求消息中的commitOffset值存储到ConsumerOffsetManager.offsetTable变量中；
			this.brokerController.getConsumerOffsetManager().commitOffset(requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getCommitOffset());
		}

		return response;
	}

	/**
	 * 发送每条消息会回调
	 */
	private List<ConsumeMessageHook> consumeMessageHookList;

	public boolean hasConsumeMessageHook() {
		return consumeMessageHookList != null && !this.consumeMessageHookList.isEmpty();
	}

	public void registerConsumeMessageHook(List<ConsumeMessageHook> sendMessageHookList) {
		this.consumeMessageHookList = sendMessageHookList;
	}

	public void executeConsumeMessageHookBefore(final ConsumeMessageContext context) {
		if (hasConsumeMessageHook()) {
			for (ConsumeMessageHook hook : this.consumeMessageHookList) {
				try {
					hook.consumeMessageBefore(context);
				} catch (Throwable e) {
				}
			}
		}
	}
}
